package cn.doitedu.rtdw.data_dashboard;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/6
 * @Desc: 学大数据，到多易教育
 * <p>
 * 截止到此刻，今天每 5分钟的pv数、uv数、会话数

 -- 测试数据：
{"user_id":3,"username":"windy","session_id":"s10","event_id":"app_launch","event_time":1680763371000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","keyword":"usb 移动固态","search_id":"sc01"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"event1","event_time":1680763372000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","res_cnt":276,"search_id":"sc01"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"page_load","event_time":1680763373000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","search_id":"sc01","item_seq":1,"item_att":"ad"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"page_load","event_time":1680763374000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","search_id":"sc01","item_seq":3,"item_att":"nt"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"event3","event_time":1680763375000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","keyword":"固态移动硬盘","search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"event4","event_time":1680763376000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763377000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"wake_up","event_time":1680763378000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"event5","event_time":1680763379000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763381000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763382000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page09","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763383000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763384000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763385000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763386000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763387000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763388000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763389000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763389100,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s11","event_id":"page_load","event_time":1680763389200,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s11","event_id":"page_load","event_time":1680763389300,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}

 -- 触发数据：
{"user_id":3,"username":"windy","session_id":"s11","event_id":"event6","event_time":1680763529300,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}

 *
 **/
public class Dash01_TrafficOverview1 {
    public static void main(String[] args) throws Exception {
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        env.setStateBackend(new HashMapStateBackend());
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka的dwd数据连接器映射表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                               "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_province STRING,                     "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     rt AS to_timestamp_ltz(event_time,3),         "
                        + "     WATERMARK FOR rt AS rt - interval '0' second  "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'earliest-offset',         "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");


        // 写一个统计sql，实现需求逻辑
        // 每 5分钟的pv、uv、新用户数，会话数
        tenv.executeSql(
                "CREATE TEMPORARY VIEW  tmp_res AS                                "
                        + "WITH  tmp AS (                                            "
                        + "   SELECT                                                 "
                        + "     *                                                    "
                        + "   FROM  dwd_kafka                                        "
                        + "   WHERE event_id = 'page_load'                           "
                        + " )                                                        "
                        + " SELECT                                                   "
                        + "  DATE_FORMAT(window_start,'yyyy-MM-dd HH:mm:ss') as window_start,"
                        + " 	count(1) as pv_amt,                                     "
                        + " 	count(distinct user_id) as uv_amt,                      "
                        + " 	count(distinct session_id) as ses_amt                   "
                        + " from TABLE(                                              "
                        + "   TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '5' MINUTE)   "
                        + " )                                                        "
                        + " GROUP BY                                                 "
                        + "   window_start,                                          "
                        + "   window_end                                             "
        );

        // 将查询结果表，转成 流
        Table resTable = tenv.from("tmp_res");
        DataStream<Dash01Bean> resStream = tenv.toDataStream(resTable, Dash01Bean.class);

        // 利用api，将结果流中的数据写入到 redis
        // redis中的数据模型： hash结构  key: dash01_pus,  value=> field:window_start, value:结果json
        // dash01_pus :
        //   2023-04-06 16:00:00 => {pv_amt:10000,uv_amt:7000,ses_amt:600}
        //   2023-04-06 16:05:00 => {pv_amt:12000,uv_amt:6000,ses_amt:700}
        //   2023-04-06 16:10:00 => {pv_amt:10400,uv_amt:6500,ses_amt:720}
        resStream.addSink(new RichSinkFunction<Dash01Bean>() {

            Jedis jedis;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 创建redis连接
                jedis = new Jedis("doitedu", 6379);
            }

            @Override
            public void invoke(Dash01Bean bean, Context context) throws Exception {
                jedis.hset("dash01_pus", bean.getWindow_start(), JSON.toJSONString(bean));
            }
        });


        env.execute();

    }
}
